extern crate tokio;

use std::pin::Pin;
use std::task::{Context, Poll};
use std::future::Future;
use tokio::net::{TcpStream, TcpListener};
use std::thread::sleep;
use std::time::Duration;
use std::thread;
use tokio::runtime::Builder;
use tokio::fs::File;
use std::path::Path;
use tokio::io::{AsyncReadExt, AsyncWriteExt, AsyncSeek};
use tokio::net::tcp::{ReadHalf, WriteHalf};
use std::borrow::BorrowMut;
use std::sync::{mpsc, Arc, Mutex};

use tokio::sync::mpsc::Receiver;
use self::tokio::io;
use self::tokio::io::ErrorKind;


//执行顺序问题
//https://blog.csdn.net/CrazyWorkers/article/details/110877643
//https://tokio-zh.github.io/document/getting-started/hello-world.html
//https://doc.rust-lang.org/book/ch16-02-message-passing.html
#[tokio::test]
async fn task_test1() {
    //主线程
    tokio::task::spawn_blocking(|| {
        //运行在一个阻塞的线程，可以看作是一个比较耗时的操作
        sleep(Duration::from_millis(1000));
        println!("1000");
    });
    tokio::task::spawn(async {
        //运行异步
        sleep(Duration::from_millis(900));
        println!("900")
    });
    //等待阻塞 道理上先输出
    tokio::task::spawn_blocking(|| {
        //运行在一个阻塞的线程，可以看作是一个比较耗时的操作
        sleep(Duration::from_millis(100));
        println!("delay 100 blocking");
    }).await;

    tokio::task::spawn(async {
        //运行异步
        sleep(Duration::from_millis(100));
        println!("delay 100 async")
    }).await;

    //最后输出
    println!("main print");
    sleep(Duration::new(1, 0));
}

//抢不到资源的情况
#[tokio::test]
async fn task_test2() {
    tokio::task::spawn(async {
        //一定最先输出
        println!("first");
    }).await;
    tokio::task::spawn(async {
        //任务所属线程可能抢占不到执行所需的资源，主线程直接退出
        println!("noop");
    });
    println!("hello");
    //切到主线程，第二输出
    sleep(Duration::from_millis(1000));
}


#[tokio::test]
async fn task_test3() {
    //主线程
    //tokio::task::spawn等价于thread::spawn的异步使用
    tokio::task::spawn(async {
        //运行在一个不阻塞的线程
        sleep(Duration::from_millis(1000));
        println!("first");
    }).await.unwrap();//使用await关键字等待阻塞线程的任务完成
    //要等待线程完成后，主线程才能执行
    println!("second");
}

//await
#[tokio::test]
async fn task_test4() {
    tokio::task::spawn_blocking(|| {
        println!("0");//使用await关键字，主线程需要等待任务执行完成才能执行
    }).await;
    tokio::task::spawn_blocking(|| {
        println!("2");//处于阻塞线程，等待主线程执行完，再执行
    });
    println!("1");
}

//yield_now
#[tokio::test]
async fn task_test5() {
    tokio::task::spawn(async {//任务时间不能超过await任务的时间，超过最后执行
        println!("1");//任务所属线程优先级较高，可以抢占到执行所需的资源
    });
    tokio::task::spawn(async {
        tokio::task::yield_now().await;//此处使用让出时间片，等待其他任务完成
        println!("2");//一定会执行，使用await会等待任务所属线程先完成，然后执行后续的操作
    }).await.unwrap();
    println!("3")
}

//block_in_place 阻塞当前线程
#[tokio::test]
async fn task_test6() {
    tokio::task::block_in_place(|| {
        println!("hi1");
    });
    tokio::task::block_in_place(|| {
        println!("hi2");
    });
    println!("hello")
}

//tokio 构造线程环境
#[tokio::test]
async fn task_test7() {
    //构造单线程tokio运行环境
    let runtime = Builder::new_multi_thread().

        enable_all().
        build().
        expect("create tokio runtime failed");
    runtime.spawn_blocking(|| {//相当于tokio::task::spawn_blocking
        //处于专属线程池中，执行耗时的操作不影响其他任务执行
        sleep(Duration::from_secs(10));
        println!("hi1");
    });
    runtime.spawn(async {//相当于tokio::task::spawn
        println!("hi2");//处于单线程中
    });
    println!("hello");
}

#[tokio::test]
async fn share_data_test() {
    let (tx, mut rx) = mpsc::channel();

    thread::spawn(move || {
        let val = String::from("hi");
        tx.send(val);
    });

    let received = rx.recv().unwrap();
    println!("Got: {}", received);
}


async fn copy(s1: &mut ReadHalf<'_>, s2: &mut WriteHalf<'_>) {
    loop {
        s1.readable().await.expect("");
        let mut buf = [0; 4096];
        let n = match s1.read(&mut buf).await {
            // socket closed
            Ok(n) if n == 0 => return,
            Ok(n) => n,
            Err(e) => {
                eprintln!("failed to read from socket; err = {:?}", e);
                return;
            }
        };
        println!("{}", n);
        s2.write_all(&buf[0..n]).await.expect("cant open");
    }
}

#[tokio::test]
async fn test_tcp_stream_connect() -> Result<(), String> {
    let mut stream = TcpStream::connect("127.0.0.1:1000").await.expect("");
    loop {
        stream.readable().await.expect("");
        let (mut r, mut w) = stream.split();
        io::copy(&mut r, &mut w).await;
    }
}

pub struct SocketStreamCopier {
    lock: Arc<Mutex<()>>,
}


impl SocketStreamCopier {
    pub fn new() -> SocketStreamCopier {
        SocketStreamCopier {
            lock: Arc::new(Mutex::new(()))
        }
    }


    pub async fn copyStream(&mut self, from: &mut TcpStream, to: &mut TcpStream) -> std::io::Result<u64> {
        self.lock.lock().unwrap();
        let mut buf = [0; 10240];
        let mut len = 0;
        loop {
            from.readable().await.expect("");
            match from.read(&mut buf).await {
                Ok(0) => {
                    return Ok(len);
                }
                Ok(bytes_read) => {
                    len += bytes_read as u64;
                    to.write(&buf[..bytes_read]).await.unwrap();
                    continue;
                }
                Err(ref e) if e.kind() == ErrorKind::Interrupted => continue,
                Err(e) => {
                    return Err(e);
                }
            }
        }
    }
}






// /// 测试端口转发失败
// #[tokio::test]
// async fn file_test() -> Result<(), Box<dyn std::error::Error>> {
//     let listener = TcpListener::bind("127.0.0.1:9988").await?;
//     loop {
//         let (mut sk, _) = listener.accept().await?;
//         tokio::spawn(async move {
//             let mut st = TcpStream::connect("127.0.0.1:1000").await.expect("cant open");
//             let sk0 = TcpStream::into_std(sk).expect("into std fail");
//             let st0 = TcpStream::into_std(st).expect("into std fail");
//             let sk1 = sk0.try_clone().expect("clone fail");
//             let st1 = st0.try_clone().expect("clone fail");
//
//
//             let t1 = thread::spawn(move || {
//
//             });
//             io::copy(&mut kr, &mut tw);
//             t1.join();
//         });
//     }
// }

#[tokio::test]
async fn rx_test() {
    let (tx, rx) = mpsc::channel();
    tokio::task::spawn_blocking(move || {
        for i in 0..100 {
            tx.send(i);
        }
    });
    loop {
        let x = rx.recv().expect("");
        println!("{}", x);
    }
}

